package com.pie4cloud.pie.bus.config;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class KafkaConfiguration {
	@Value(value = "${spring.kafka.consumer.auto-commit-interval}")
	private int autoCommitInterval;
	@Value(value = "${spring.kafka.bootstrap-servers}")
	private String bootstrap;
	@Value(value = "${spring.kafka.consumer.auto-offset-reset}")
	private String autoOffsetReset;
	@Value(value = "${spring.kafka.consumer.enable-auto-commit}")
	private Boolean enableAutoCommit;
	@Value(value = "${spring.kafka.consumer.key-deserializer}")
	private String keyDeserializer;
	@Value(value = "${spring.kafka.consumer.value-deserializer}")
	private String valueDeserializer;

	public Map<String, Object> consumerConfigsEarliest() {
		Map<String, Object> props = new HashMap<String, Object>();
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
		props.put("key.deserializer", keyDeserializer);
		props.put("value.deserializer", valueDeserializer);
		//props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,autoCommitInterval);
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
		return props;
	}
	public Map<String, Object> consumerConfigsLatest() {
		Map<String, Object> props = new HashMap<String, Object>();
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
		props.put("key.deserializer", keyDeserializer);
		props.put("value.deserializer", valueDeserializer);
		//props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,autoCommitInterval);
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
		return props;
	}
	@Bean
	public KafkaListenerContainerFactory<?> batchFactoryLatest() {
		ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
		factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigsLatest()));
		// 设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
		factory.setBatchListener(false);
		factory.setConcurrency(1);
		factory.setMissingTopicsFatal(false);
		factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
		factory.getContainerProperties().setPollTimeout(30000);
		return factory;
	}

	@Bean
	public KafkaListenerContainerFactory<?> batchFactoryEarliest() {
		ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
		factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigsEarliest()));
		// 设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
		factory.setBatchListener(false);
		factory.setConcurrency(1);
		factory.setMissingTopicsFatal(false);
		factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
		factory.getContainerProperties().setPollTimeout(30000);
		return factory;
	}
	@Bean
	public KafkaAdmin kafkaAdmin() {
		Map<String, Object> props = new HashMap<>();
		//配置Kafka实例的连接地址
		props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
		KafkaAdmin admin = new KafkaAdmin(props);
		return admin;
	}

	@Bean
	public AdminClient adminClient() {
		return AdminClient.create(kafkaAdmin().getConfigurationProperties());
	}

}
